package com.cctc.juc.source.Athread.threadpool;

import java.util.HashSet;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * [ThreadPoolExecutor 线程池] （核心）实现类。
 * 要求：[（熟练）手撕] ThreadPoolExecutor 线程池的核心逻辑实现代码。
 * ThreadPoolExecutor 线程池核心逻辑内部实现原理（熟练表述）：
 * ThreadPoolExecutor 线程池的核心逻辑就是线程池中的工作线程不断从任务队列中获取任务去执行，其中工作线程 Worker 是线程池中的一个内部类，该类继承自 AQS，并实现了 Runnbale 接口，且内部依赖于一个任务执行器线程实例，实现了不断的从任务队列中获取任务并以线程安全的方式互斥执行。
 * 1）当用户向线程池中提交一个任务时，线程池会根据其配置参数以及当前池中（核心 & 非核心）工作线程的数量进行动态伸缩，例如当（核心）工作线程数量 count < corePoolSize 时，会新建一个（核心）工作线程，将其放入到线程池中，执行新提交的任务；否则，就会将新提交的任务加入到（有界）任务队列中进行排队；当（有界）任务队列满时，线程池又会再次创建一个（非核心）工作线程，将其放入到线程池中，执行新提交的任务；当（核心 & 非核心）工作线程数量 > maximumPoolSize 时，新提交的任务将会被线程池采用配置好的策略进行拒绝；此外，被加入到线程池中的工作线程会启动 “自旋” 执行任务流程，不断地从任务队列中获取任务，然后以线程安全的方式互斥执行；当线程池不繁忙时，（非核心）线程空闲时间一旦超过 keepAliveTime，就会被线程池所回收，以节省资源；最终，线程池中工作线程的数量会逐渐趋于 corePoolSize，达到相对平衡状态。
 * 2）当线程池中的任务队列不为空但无任何工作线程可用，或者任务队列为空且所有任何工作线程都被终止时，或者用户显式调用了 shutdown()、shutdownNow()、terminate() 方法，线程池都将会进行终止状态。
 * 总而言之，ThreadPoolExecutor 线程池通过非常优秀的设计，可以实现动态调整工作线程的数量，合理复用线程资源，去处理用户提交的任务，以达到理想的效果。
 */
public class DThreadPoolExecutor extends AbstractExecutorService {
    private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
        final Thread executor;
        Runnable firstTask;
        volatile long completedTasks;

        public Worker(Runnable firstTask) {
            setState(-1);
            this.firstTask = firstTask;
            executor = new Thread(this);
        }

        @Override
        public void run() {
            runWorker(this);
        }

        void interruptIfStarted() {
            Thread exec;
            if (getState() >= 0 && (exec = executor) != null && !exec.isInterrupted()) {
                try {
                    exec.interrupt();
                } catch (SecurityException ignored) {
                }
            }
        }

        public void lock() {
            acquire(1);
        }

        public void tryLock() {
            tryAcquire(1);
        }

        public void unlock() {
            release(1);
        }

        public boolean isLocked() {
            return isHeldExclusively();
        }

        @Override
        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }
    }

    // Worker 工作线程集合：将 Worker 工作线程以线程安全的方式加入到线程池中；
    private final HashSet<Worker> workers = new HashSet<>();
    private final ReentrantLock mainLock = new ReentrantLock();
    private final Condition termination = mainLock.newCondition();
    // 核心线程数、最大线程数
    private volatile int corePoolSize, maximumPoolSize;
    //（非核心）线程空闲时间
    private volatile long keepAliveTime;
    // Worker 工作线程实例工厂
    private volatile ThreadFactory threadFactory;
    // 任务队列
    private final BlockingQueue<Runnable> workQueue = new LinkedBlockingDeque<>();
    // 线程池总完成任务数
    private long completedTaskCount;
    // 新提交任务拒绝策略
    private volatile RejectedExecutionHandler handler;

    public DThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                               long keepAliveTime, TimeUnit unit,
                               BlockingQueue<Runnable> workQueue,
                               ThreadFactory threadFactory,
                               RejectedExecutionHandler handler) {
        // ... 初始化线程池 ...
    }

    /**
     * 向线程池中提交一个任务。
     * 1）如果当前线程池中的（核心）工作线程数量 count < corePoolSize，则新建一个（核心）工作线程，并将其加入到线程池中执行任务；
     * 2）否则，将新提交的任务放入任务队列；
     * 3）如果任务队列已满，则再新建一个（非核心）工作线程，并将其加入到线程池中执行任务；
     * 4）如果当前线程池中的（核心 & 非核心）工作线程数量 count > maximumPoolSize，则拒绝新提交的任务；
     * 此外，当一个 Worker 工作线程执行完当前任务后，它会自动从任务队列中取下一个任务来执行。
     * 当一个 Worker（非核心）工作线程无事可做，超过一定的时间（keepAliveTime）时，线程池会进行判断：
     * - 如果当前运行的 Worker 工作线程数大于 corePoolSize ，那么这个 Worker（非核心）工作线程就被终止。
     * 所以，当线程池不繁忙时，Worker 工作线程的数量最终会收缩到 corePoolSize 的大小。
     * 最后，如果线程池的任务队列非空但没有可用的 Worker 工作线程，或者任务队列为空且所有 Worker 工作线程都已回收（corePoolSize = 0 || allowCoreThreadTimeOut = true），线程池将会被终止。
     */
    @Override
    public void execute(Runnable command) {
        if (command == null) throw new NullPointerException();
        // 获取线程池的 [3] 状态 state 和 [29] 线程数量 count
        int c = ctl.get();
        // 1）如果线程池中的 Worker 工作线程数量 count < corePoolSize，则新建一个 Worker（核心）工作线程，将其加入到线程池中，执行新提交的任务；
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true)) return;
            c = ctl.get();
        }
        // 2）否则，将新提交的任务加入到任务队列；
        if (isRunning(c) && workQueue.offer(command)) {
            // 重新检查线程池的 [3] 状态 state 和 [29] 线程数量 count
            int recheck = ctl.get();
            // 如果线程池的状态 state != RUNNING，则将新提交的任务从任务队列中移除出去
            if (!isRunning(recheck) && remove(command)) {
                // 执行拒绝策略
                reject(command);
            }
            // 如果线程数量 count == 0，则向线程池中添加一个（非核心）线程，执行任务，保证线程池正常运行。
            else if (workerCountOf(recheck) == 0) {
                addWorker(null, false);
            }
        }
        // 3）如果（有界）任务队列 workQueue 已满，则向线程池中添加一个 Worker（非核心）工作线程，执行新提交的任务；
        else if (!addWorker(command, false)) {
            // 4）如果线程池中的 Worker（核心 & 非核心）工作线程数量 count > maximumPoolSize，则执行拒绝策略；
            reject(command);
        }
    }

    /**
     * 向线程池中添加一个工作线程，并启动它的任务执行器，执行目标任务。
     * 1）根据新提交的任务创建一个工作线程；
     * 2）将工作线程以线程安全的方式加入到线程池中；
     * 3）启动工作线程的任务执行器，执行目标任务。
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        boolean workerAdded = false, workerStarted = false;
        Worker worker = null;
        try {
            //  1）根据新提交的任务创建一个工作线程；
            worker = new Worker(firstTask);
            final Thread executor = worker.executor;
            if (executor != null) {
                // 2）将工作线程以线程安全的方式加入到线程池中；
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    workers.add(worker);
                    workerAdded = true;
                } finally {
                    mainLock.unlock();
                }
                // 3）启动工作线程的任务执行器，执行目标任务。
                if (workerAdded) {
                    executor.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (!workerStarted) addWorkerFailed(worker);
        }
        return workerStarted;
    }

    private void addWorkerFailed(Worker worker) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (worker != null) {
                workers.remove(worker);
            }
        } finally {
            mainLock.unlock();
        }
    }

    /**
     * 运行任务执行器，启动 “自旋” 执行任务流程。
     * 1）获取当前任务执行器；
     * 2）更新锁状态 -1 -> 0，标记当前工作线程完成初始化，且已启动，允许被中断；
     * 3）启动 “自旋” 执行任务流程：获取新提交的任务，或者不断从线程池的任务队列中获取任务，并以线程安全的方式互斥执行；
     * 3.1）加锁（0 -> 1）；
     * 3.2）主动检查是否需要中断任务执行器的 “自旋” 执行任务流程；
     * 3.3）调用 beforeExecute() 钩子函数，运行任务执行前的处理逻辑；
     * 3.4）调用 task.run()，执行目标任务；
     * 3.5）调用 afterExecute() 钩子函数，运行任务执行后的处理逻辑；
     * 3.6）解锁（1 -> 0），并累加任务完成数 + 1；
     * 4）工作线程终止前，累加线程池的总任务完成数，然后从线程池中移除工作线程。
     */
    final void runWorker(Worker worker) {
        // 1）获取当前任务执行器；
        Thread executor = Thread.currentThread();
        Runnable task = worker.firstTask;
        worker.firstTask = null;
        // 2）更新锁状态 -1 -> 0，标记当前工作线程完成初始化，且已启动，允许被中断；
        worker.unlock();
        try {
            //  3）启动 “自旋” 执行任务流程：获取新提交的任务，或者不断从线程池的任务队列中获取任务，并以线程安全的方式互斥执行；
            while (task != null || (task = getTask()) != null) {
                // 3.1）加锁（0 -> 1）；
                worker.lock();
                // 3.2）主动检查是否需要中断任务执行器的 “自旋” 执行任务流程；
                if (Thread.interrupted() && !executor.isInterrupted()) executor.interrupt();
                try {
                    // 3.3）调用 beforeExecute() 钩子函数，运行任务执行前的处理逻辑；
                    beforeExecute(executor, task);
                    Throwable thrown = null;
                    try {
                        // 3.4）调用 task.run() 方法，执行目标任务；
                        task.run();
                    } catch (Exception e) {
                        thrown = e;
                        throw e;
                    } finally {
                        // 3.5）调用 afterExecute() 钩子函数，运行任务执行后的处理逻辑；
                        afterExecute(task, thrown);
                    }
                } finally {
                    // 3.6）解锁（1 -> 0），并累加任务完成数 + 1；
                    task = null;
                    worker.completedTasks++;
                    worker.unlock();
                }
            }
        } finally {
            // 4）工作线程终止前，累加线程池的总任务完成数，然后从线程池中移除工作线程。
            processWorkerExit(worker);
        }
    }

    private Runnable getTask() {
        for (; ; ) {
            if (workQueue.isEmpty()) return null;
            try {
                Runnable task = workQueue.take();
                if (task != null) return task;
            } catch (InterruptedException retry) {
            }
        }
    }

    protected void beforeExecute(Thread t, Runnable r) {
        // 任务执行前的钩子函数，交由子类实现处理逻辑。
    }

    protected void afterExecute(Runnable r, Throwable t) {
        // 任务执行后的钩子函数，交由子类实现处理逻辑。
    }

    private void processWorkerExit(Worker worker) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += worker.completedTasks;
            workers.remove(worker);
        } finally {
            mainLock.unlock();
        }
    }

    @Override
    public void shutdown() {

    }

    @Override
    public List<Runnable> shutdownNow() {
        return null;
    }

    @Override
    public boolean isShutdown() {
        return false;
    }

    @Override
    public boolean isTerminated() {
        return false;
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return false;
    }

}
